package videoInfo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author legolas
 * @date 2020/3/14 下午5:09
 * 1.基于主播进行统计，统计每个主播在当天收到的总金币数量，总观看pv，关注量，总开播时长
 * <p>
 * 分析
 * 1.为了方便统计主播的指标数据，最好把这些字段整合到一个对象中
 * 2.由于需要以主播维度进行数据的聚合，所以以主播ID作为key
 * 3.Map节点的<k2,v2>为<Text,自定义Writable>
 * 4.需要做聚合操作，所以有reduce阶段
 */
public class VideoInfoJob {

    public static void main(String[] args) {
        try {
            if (args.length != 2) {
                //如果参数不够，程序直接退出
                System.out.print("请指定输入路径和输出路径！");
                System.exit(100);
            }
            //创建Job需要的配置参数
            Configuration conf = new Configuration();
            //创建一个job
            Job job = Job.getInstance(conf);

            //注意：这一行必须设置，否则在集群中执行找不到WordCountJob这个类
            job.setJarByClass(VideoInfoJob.class);

            //指定输入路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            //指定map相关代码
            job.setMapperClass(VideoInfoMap.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(VideoInfoWritable.class);

            //指定reduce相关代码
            job.setReducerClass(VideoInfoReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(VideoInfoWritable.class);

            //提交job
            job.waitForCompletion(true);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
